package com.world.rickcloudendpoint.demo.local.consumer;

import com.world.rickcloudendpoint.demo.model.DemoMessage;
import com.world.rickcloudendpoint.demo.MsgConsumer;
import com.world.rickcloudendpoint.demo.QueueConstantKey;
import com.world.rickcloudendpoint.demo.RedisService;
import com.world.rickcloudendpoint.demo.local.LocalQueueFactory;
import lombok.extern.log4j.Log4j2;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.AbstractQueue;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

/**
 * @description:
 * @author: jlsong
 * @date: 2022/06/03 20:49:16
 * @version: 1.0
 */
@Component
@Log4j2
public class BusinessEightConsumer implements MsgConsumer {
    @Resource
    LocalQueueFactory queueFactory;

    final
    RedisService redisService;

    public BusinessEightConsumer(RedisService redisService) {
        this.redisService = redisService;
    }

    @Override
    public void onMessage(Object message) {
        String cacheKeyPrefix = QueueConstantKey.CACHE_PREFIX.concat(":").concat(QueueConstantKey.REAL_TIME_EIGHT).concat(":");
        try {
            // 解析消息并放入本地队列进行处理，缓存到内存中
            if (message != null && message instanceof List) {
                List<DemoMessage> collects = (List<DemoMessage>) message;
                if (collects != null && collects.size() > 0) {
                    Map<String, List<DemoMessage>> groupByStationIdAndTargetId = collects.stream().collect(Collectors.groupingBy(r -> r.getStationId() + ":" + r.getTargetId()));

                    groupByStationIdAndTargetId.values().stream().forEach(r -> {
                        // 判断是否存在
                        List<DemoMessage> sortRecords = r.stream().sorted(Comparator.comparing(DemoMessage::getLatestTime)).collect(Collectors.toList());
                        if (sortRecords != null && sortRecords.size() > 0) {
                            DemoMessage latest = sortRecords.get(sortRecords.size() - 1);
                            // 目标的redis中的key
                            String cacheKey = cacheKeyPrefix.concat(latest.getStationId() + ":" + latest.getTargetId());
                            redisService.set(cacheKey, latest, QueueConstantKey.EXPIRE_TIME);
                        }
                    });

                    // 将消息转入到本地持久化队列中
                    // 持久化队列 名称
                    String persistedQueueName = QueueConstantKey.CACHE_PREFIX.concat(":").concat(QueueConstantKey.PERSISTED_EIGHT);
                    AbstractQueue queue = queueFactory.getQueue(persistedQueueName);
                    if (queue != null) {
                        LinkedBlockingQueue blockingQueue = (LinkedBlockingQueue) queue;
                        // 判断队列是否已满
                        if (blockingQueue.remainingCapacity() < collects.size()) {
                            log.info("queue[{}] is full,can not add item in queue.", persistedQueueName);
                            return;
                        }
                        blockingQueue.put(collects);
                    }
                }
            }

        } catch (Exception e) {
            log.error("business[{}] consumed failed,the reason is {} ", cacheKeyPrefix, e.getMessage(), e);
        }
    }

    @Override
    public void onError(Object message, Exception e) {
        log.error("业务逻辑8消费者=>存储消息发生错误", message, e);
    }
}
